package com.micro.mq.controller;

import com.micro.common.dto.MqDto;
import com.micro.common.vo.R;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;

/**
 * @author pan.shun
 * @since 2022/7/19 15:35
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${app.topic.foo2}")
    private String topic;

    @Value("${app.topic.foo}")
    private String topic2;

    /**
     * Kafka生产者延时消费
     */
    @GetMapping("/saySomeThing/{word}")
    public R sendMsg(@PathVariable String word) {
        kafkaTemplate.send(topic2, word);
        return R.success();
    }

    @PostMapping("/sayTopic")
    public R sendMsg(@RequestBody MqDto mqDto) {

        if (StringUtils.isEmpty(mqDto.getTopic())) {
            return R.error("[Topic] 不能为空");
        }
        if (StringUtils.isEmpty(mqDto.getMessage())) {
            return R.error("[Message] 不能为空");
        }

        kafkaTemplate.send(mqDto.getTopic(), mqDto.getMessage());
        return R.success();
    }


    /**
     * Kafka生产者批量发送消息
     */
    @GetMapping("/sayBatchSomeThing/{word}")
    public R sendMoreMsg(@PathVariable String word) {

        for (int i = 0; i < 20; i++) {
            kafkaTemplate.send(topic, word);
        }
        return R.success();
    }
}
